ποΈGitΠ―ΡΠ°ποΈ
Commit be320ec622b8cdf951c60e55ec26620985b0eabc
Parents : 1b0b00c
Author : James Rich <2199651+jamesarich@users.noreply.github.com>
Signature : Signature validation error
Date : 2026-07-28T20:09:23-05:00
Committer : GitHub <noreply@github.com>
Date : 2026-07-28T20:09:23-05:00
fix(mqtt): stop reporting transport failures as credential rejections (#6506)
Co-authored-by: Claude Fable 5 <noreply@anthropic.com>
Changes
5 files changed, 194 insertions(+), 11 deletions(-)
Diff
diff --git a/.skills/compose-ui/strings-index.txt b/.skills/compose-ui/strings-index.txt
index f4325fa808..b87add6771 100644
--- a/.skills/compose-ui/strings-index.txt
+++ b/.skills/compose-ui/strings-index.txt
@@ -1057,6 +1057,10 @@ mpwrd_os
mqtt
mqtt_config
mqtt_enabled
+mqtt_error_connection_lost
+mqtt_error_credentials_rejected
+mqtt_error_proxy_failed
+mqtt_error_rejected
mqtt_probe_dns_failure
mqtt_probe_other_failure
mqtt_probe_rejected
diff --git a/core/data/src/commonMain/kotlin/org/meshtastic/core/data/manager/MqttManagerImpl.kt b/core/data/src/commonMain/kotlin/org/meshtastic/core/data/manager/MqttManagerImpl.kt
index 68e6b9fa43..520128dfb5 100644
--- a/core/data/src/commonMain/kotlin/org/meshtastic/core/data/manager/MqttManagerImpl.kt
+++ b/core/data/src/commonMain/kotlin/org/meshtastic/core/data/manager/MqttManagerImpl.kt
@@ -31,15 +31,25 @@ import kotlinx.coroutines.flow.onEach
import kotlinx.coroutines.flow.stateIn
import org.koin.core.annotation.Named
import org.koin.core.annotation.Single
+import org.meshtastic.core.common.util.safeCatchingAll
import org.meshtastic.core.model.MqttConnectionState
import org.meshtastic.core.model.MqttProbeStatus
import org.meshtastic.core.network.repository.MQTTRepository
+import org.meshtastic.core.network.repository.MQTT_KEEPALIVE_SECONDS
+import org.meshtastic.core.network.repository.isCredentialRejection
import org.meshtastic.core.network.repository.mqttTlsConfig
import org.meshtastic.core.network.repository.resolveEndpoint
import org.meshtastic.core.repository.MqttManager
import org.meshtastic.core.repository.NodeRepository
import org.meshtastic.core.repository.PacketHandler
import org.meshtastic.core.repository.ServiceStateWriter
+import org.meshtastic.core.resources.Res
+import org.meshtastic.core.resources.getStringSuspend
+import org.meshtastic.core.resources.mqtt_error_connection_lost
+import org.meshtastic.core.resources.mqtt_error_credentials_rejected
+import org.meshtastic.core.resources.mqtt_error_proxy_failed
+import org.meshtastic.core.resources.mqtt_error_rejected
+import org.meshtastic.core.resources.unknown
import org.meshtastic.mqtt.ConnectionState
import org.meshtastic.mqtt.MqttClient
import org.meshtastic.mqtt.MqttException
@@ -80,12 +90,26 @@ class MqttManagerImpl(
.onEach { message -> packetHandler.sendToRadio(ToRadio(mqttClientProxyMessage = message)) }
.catch { throwable ->
_proxyActive.value = false
+ // safeCatchingAll swallows the Skiko ExceptionInInitializerError that
+ // compose-resources raises on headless JVM tests; production resolves the
+ // localized string and the error is still surfaced either way.
val message =
- when (throwable) {
- is MqttException.ConnectionRejected -> "MQTT: connection rejected (check credentials)"
- is MqttException.ConnectionLost -> "MQTT: connection lost"
- else -> "MQTT proxy failed: ${throwable.message}"
+ safeCatchingAll {
+ when {
+ throwable is MqttException.ConnectionRejected &&
+ throwable.isCredentialRejection() ->
+ getStringSuspend(Res.string.mqtt_error_credentials_rejected)
+
+ throwable is MqttException.ConnectionRejected ->
+ getStringSuspend(Res.string.mqtt_error_rejected, throwable.detail())
+
+ throwable is MqttException.ConnectionLost ->
+ getStringSuspend(Res.string.mqtt_error_connection_lost)
+
+ else -> getStringSuspend(Res.string.mqtt_error_proxy_failed, throwable.detail())
+ }
}
+ .getOrDefault("")
serviceStateWriter.setErrorMessage(text = message, severity = Severity.Warn)
}
.launchIn(scope)
@@ -144,6 +168,9 @@ class MqttManagerImpl(
// including its scoped private-CA trust hook β otherwise a probe would fail where a connect succeeds.
val tls = mqttTlsConfig()
transportFactory = TcpTransportFactory(tls) + WebSocketTransportFactory(tls)
+ // Mirror the live client's keepalive too: the library default is 0 (no keepalive),
+ // which some brokers reject β misleadingly, as CLIENT_IDENTIFIER_NOT_VALID.
+ keepAliveSeconds = MQTT_KEEPALIVE_SECONDS
// Per-connection random suffix: myId identifies the node (and is null β
// "unknown" before the node record loads), so two probes can collide on one
// client-id and evict each other (SESSION_TAKEN_OVER). See MQTTRepositoryImpl.
@@ -188,3 +215,6 @@ class MqttManagerImpl(
is ProbeResult.Other -> MqttProbeStatus.Other(message = cause.message)
}
}
+
+/** Failure detail for a user-facing message; a throwable without a message still needs a placeholder to substitute. */
+private suspend fun Throwable.detail(): String = message ?: getStringSuspend(Res.string.unknown)
diff --git a/core/network/src/commonMain/kotlin/org/meshtastic/core/network/repository/MQTTRepositoryImpl.kt b/core/network/src/commonMain/kotlin/org/meshtastic/core/network/repository/MQTTRepositoryImpl.kt
index 83c01c6b1e..fa76e3a70e 100644
--- a/core/network/src/commonMain/kotlin/org/meshtastic/core/network/repository/MQTTRepositoryImpl.kt
+++ b/core/network/src/commonMain/kotlin/org/meshtastic/core/network/repository/MQTTRepositoryImpl.kt
@@ -52,6 +52,7 @@ import org.meshtastic.mqtt.MqttException
import org.meshtastic.mqtt.MqttLogLevel
import org.meshtastic.mqtt.MqttMessage
import org.meshtastic.mqtt.QoS
+import org.meshtastic.mqtt.ReasonCode
import org.meshtastic.mqtt.packet.Subscription
import org.meshtastic.mqtt.plus
import org.meshtastic.mqtt.transport.tcp.TcpTransportFactory
@@ -199,17 +200,17 @@ class MQTTRepositoryImpl(
}
Logger.i { "MQTT connected and subscribed" }
}
+ val failure = result.exceptionOrNull()
when {
result.isSuccess -> return@launch
- result.exceptionOrNull() is MqttException.ConnectionRejected -> {
- Logger.e(result.exceptionOrNull()) { "MQTT connection rejected (unrecoverable), stopping" }
- close(result.exceptionOrNull()!!)
+ failure is MqttException.ConnectionRejected && failure.isCredentialRejection() -> {
+ Logger.e(failure) { "MQTT connection rejected (unrecoverable), stopping" }
+ close(failure)
return@launch
}
else -> {
- val failure = result.exceptionOrNull()
// Broker- and network-side failures are what this retry loop exists to absorb β an
// unreachable host, a TLS problem, a dropped connection, or a broker that violates the
// MQTT 5 spec (e.g. the topic-alias limit). None are defects in this app, and reporting
@@ -320,6 +321,24 @@ class MQTTRepositoryImpl(
}
}
+/**
+ * `true` only for CONNACK reason codes where retrying can never help β the broker examined our credentials or client
+ * identity and refused them. The library also wraps transport-level connect failures (timeout, TLS, socket EOF) as
+ * [MqttException.ConnectionRejected] with [ReasonCode.UNSPECIFIED_ERROR]; those are transient and must stay in the
+ * retry loop, not permanently stop the proxy with a "check credentials" dialog. Public (not internal) so
+ * `MqttManagerImpl` in `:core:data` can phrase its user-facing error from the same classification.
+ */
+fun MqttException.ConnectionRejected.isCredentialRejection(): Boolean = when (reasonCode) {
+ ReasonCode.BAD_USER_NAME_OR_PASSWORD,
+ ReasonCode.NOT_AUTHORIZED,
+ ReasonCode.BAD_AUTHENTICATION_METHOD,
+ ReasonCode.CLIENT_IDENTIFIER_NOT_VALID,
+ ReasonCode.BANNED,
+ -> true
+
+ else -> false
+}
+
/**
* `true` when an MQTT connect/subscribe failure is one the retry loop is designed to absorb, rather than a defect worth
* reporting to Crashlytics/Datadog.
@@ -387,8 +406,9 @@ private fun defaultMqttClientFactory(setup: MqttClientSetup): MqttClientSession
transportFactory = TcpTransportFactory(tls) + WebSocketTransportFactory(tls)
keepAliveSeconds = MQTT_KEEPALIVE_SECONDS
autoReconnect = true
- username = setup.mqttConfig?.username
- setup.mqttConfig?.password?.let { password(it) }
+ val (user, pass) = effectiveCredentials(setup.mqttConfig)
+ username = user
+ pass?.let { password(it) }
logger = KermitMqttLogger()
// WARN for production: the library emits endpoint addresses and topic strings at
// INFO level. WARN messages (reconnect, timeout, retry) contain no PII and are
@@ -397,7 +417,9 @@ private fun defaultMqttClientFactory(setup: MqttClientSetup): MqttClientSession
},
)
-private const val MQTT_KEEPALIVE_SECONDS = 30
+// Public (not internal/private) so MqttManagerImpl's probe in :core:data can mirror the live client β
+// some brokers reject a keepalive-0 CONNECT (misleadingly, as CLIENT_IDENTIFIER_NOT_VALID).
+const val MQTT_KEEPALIVE_SECONDS = 30
private const val MQTT_PORT_PLAIN = 1883
private const val MQTT_PORT_TLS = 8883
@@ -426,6 +448,26 @@ fun resolveEndpoint(rawAddress: String, tlsEnabled: Boolean): MqttEndpoint = if
private const val DEFAULT_PUBLIC_SERVER = "mqtt.meshtastic.org"
+// The public broker's well-known credentials, same values as the firmware's Default.h.
+private const val DEFAULT_MQTT_USERNAME = "meshdev"
+private const val DEFAULT_MQTT_PASSWORD = "large4cats"
+
+/**
+ * Mirrors the firmware's `PubSubConfig` rule: an empty `address` means "the public broker with its well-known
+ * credentials", substituting username and password together with the server β the stored username/password are ignored
+ * in that case, whatever they contain. Substituting only the address (as this repository does for the endpoint) while
+ * passing the stored empty credentials through makes the proxy connect anonymously, which the public broker rejects
+ * with BAD_USER_NAME_OR_PASSWORD β surfaced to the user as a bogus "check credentials" error on configs the firmware
+ * itself connects with happily. Empty-address configs occur in the wild: lockdown-enabled firmware hands
+ * unauthenticated clients a zeroed MQTTConfig.
+ */
+internal fun effectiveCredentials(config: ModuleConfig.MQTTConfig?): Pair<String?, String?> =
+ if (config?.address.isNullOrEmpty()) {
+ DEFAULT_MQTT_USERNAME to DEFAULT_MQTT_PASSWORD
+ } else {
+ config?.username to config?.password
+ }
+
fun effectiveTlsEnabled(address: String, tlsEnabled: Boolean): Boolean =
tlsEnabled || extractHost(address).equals(DEFAULT_PUBLIC_SERVER, ignoreCase = true)
diff --git a/core/network/src/commonTest/kotlin/org/meshtastic/core/network/repository/MQTTRepositoryImplTest.kt b/core/network/src/commonTest/kotlin/org/meshtastic/core/network/repository/MQTTRepositoryImplTest.kt
index a3b3b53a29..50caf893f9 100644
--- a/core/network/src/commonTest/kotlin/org/meshtastic/core/network/repository/MQTTRepositoryImplTest.kt
+++ b/core/network/src/commonTest/kotlin/org/meshtastic/core/network/repository/MQTTRepositoryImplTest.kt
@@ -37,6 +37,7 @@ import kotlinx.coroutines.test.runTest
import kotlinx.serialization.json.Json
import okio.ByteString.Companion.toByteString
import org.meshtastic.core.common.BuildConfigProvider
+import org.meshtastic.core.common.util.safeCatching
import org.meshtastic.core.di.CoroutineDispatchers
import org.meshtastic.core.model.MqttJsonPayload
import org.meshtastic.core.testing.FakeNodeRepository
@@ -184,6 +185,40 @@ class MQTTRepositoryImplTest {
// endregion
+ // region effectiveCredentials β firmware-parity credential defaulting.
+
+ @Test
+ fun `empty address substitutes the public broker's well-known credentials`() {
+ // Mirrors firmware PubSubConfig: lockdown-redacted (zeroed) configs must not connect anonymously.
+ val creds = effectiveCredentials(ModuleConfig.MQTTConfig(address = "", username = "", password = ""))
+ assertEquals("meshdev" to "large4cats", creds)
+ }
+
+ @Test
+ fun `null config substitutes the public broker's well-known credentials`() {
+ assertEquals("meshdev" to "large4cats", effectiveCredentials(null))
+ }
+
+ @Test
+ fun `empty address ignores stored credentials entirely - firmware parity`() {
+ val creds = effectiveCredentials(ModuleConfig.MQTTConfig(address = "", username = "custom", password = "pw"))
+ assertEquals("meshdev" to "large4cats", creds)
+ }
+
+ @Test
+ fun `explicit address uses the stored credentials as-is`() {
+ val config = ModuleConfig.MQTTConfig(address = "broker.example.com", username = "user", password = "pass")
+ assertEquals("user" to "pass", effectiveCredentials(config))
+ }
+
+ @Test
+ fun `explicit default server address uses the stored credentials as-is - firmware parity`() {
+ val config = ModuleConfig.MQTTConfig(address = "mqtt.meshtastic.org", username = "user", password = "pass")
+ assertEquals("user" to "pass", effectiveCredentials(config))
+ }
+
+ // endregion
+
// region extractHost β address canonicalization tests.
@Test
@@ -337,6 +372,74 @@ class MQTTRepositoryImplTest {
runCurrent()
}
+ @Test
+ fun `transport failure wrapped as ConnectionRejected retries instead of stopping the proxy`() = runTest {
+ // The MQTT library wraps ANY connect failure β timeout, TLS, socket EOF β as
+ // ConnectionRejected with UNSPECIFIED_ERROR. Treating those as unrecoverable stopped
+ // the proxy permanently and showed users a bogus "check credentials" dialog.
+ val harness = createHarness()
+ harness.client.failConnectWith(
+ MqttException.ConnectionRejected(ReasonCode.UNSPECIFIED_ERROR, "Connection failed: Connection timed out"),
+ )
+
+ val collector = startProxyCollection(harness.repository)
+ runCurrent()
+ assertEquals(1, harness.client.connectCalls.size)
+
+ advanceTimeBy(1_000)
+ runCurrent()
+ assertEquals(2, harness.client.connectCalls.size)
+ assertEquals(1, harness.client.subscribeCalls.size)
+
+ collector.cancelAndJoin()
+ runCurrent()
+ }
+
+ @Test
+ fun `credential rejection stops the proxy permanently`() = runTest {
+ val harness = createHarness()
+ harness.client.failConnectWith(
+ MqttException.ConnectionRejected(ReasonCode.BAD_USER_NAME_OR_PASSWORD, "Connection refused"),
+ )
+
+ val outcome = backgroundScope.async { safeCatching { harness.repository.proxyMessageFlow.collect {} } }
+ runCurrent()
+ advanceTimeBy(60_000)
+ runCurrent()
+
+ assertEquals(1, harness.client.connectCalls.size)
+ assertIs<MqttException.ConnectionRejected>(outcome.await().exceptionOrNull())
+ }
+
+ @Test
+ fun `only credential and identity reason codes classify as credential rejections`() {
+ val fatal =
+ listOf(
+ ReasonCode.BAD_USER_NAME_OR_PASSWORD,
+ ReasonCode.NOT_AUTHORIZED,
+ ReasonCode.BAD_AUTHENTICATION_METHOD,
+ ReasonCode.CLIENT_IDENTIFIER_NOT_VALID,
+ ReasonCode.BANNED,
+ )
+ val transient =
+ listOf(
+ ReasonCode.UNSPECIFIED_ERROR,
+ ReasonCode.SERVER_UNAVAILABLE,
+ ReasonCode.SERVER_BUSY,
+ ReasonCode.CONNECTION_RATE_EXCEEDED,
+ )
+
+ fatal.forEach { code ->
+ assertTrue(MqttException.ConnectionRejected(code, "x").isCredentialRejection(), "expected fatal: $code")
+ }
+ transient.forEach { code ->
+ assertFalse(
+ MqttException.ConnectionRejected(code, "x").isCredentialRejection(),
+ "expected transient: $code",
+ )
+ }
+ }
+
@Test
fun `subscription failures trigger reconnect retry`() = runTest {
val harness = createHarness()
diff --git a/core/resources/src/commonMain/composeResources/values/strings.xml b/core/resources/src/commonMain/composeResources/values/strings.xml
index 6d90f3b4e6..4be92ff63b 100644
--- a/core/resources/src/commonMain/composeResources/values/strings.xml
+++ b/core/resources/src/commonMain/composeResources/values/strings.xml
@@ -1090,6 +1090,10 @@
<string name="mqtt">MQTT</string>
<string name="mqtt_config">MQTT Config</string>
<string name="mqtt_enabled">MQTT enabled</string>
+ <string name="mqtt_error_connection_lost">MQTT: connection lost</string>
+ <string name="mqtt_error_credentials_rejected">MQTT: connection rejected (check credentials)</string>
+ <string name="mqtt_error_proxy_failed">MQTT proxy failed: %1$s</string>
+ <string name="mqtt_error_rejected">MQTT: connection rejected: %1$s</string>
<string name="mqtt_probe_dns_failure">Host not found</string>
<string name="mqtt_probe_other_failure">Connection failed</string>
<string name="mqtt_probe_rejected">Broker rejected: %1$s</string>
Served by rngit 1.5.2 - Generated in 0.22s